package com.zookeeper.lock;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.CreateMode;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Author: dong peng
 * Date: 2018/1/18
 * Time: 18:03
 *
 * 实现思路：
 * 1. 在某个路径下创建临时节点
 * 2. 检查，获取所有的临时节点，序号最小的获的锁
 * 3. 成功则删除该节点，通知相应的订阅者。 失败则再次抢锁
 */


public class DistributeLock {

    private ZkClient zkClient;

    // 根节点
    private String dirPrefix = "/tmp";


    public void getLock() {
        ini();
        createRootNode();
        createEmpNode();
    }

    private void  ini() {
        zkClient = new ZkClient("127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183", 3000);
    }

    private void createRootNode() {
        if(StringUtils.isNotEmpty(dirPrefix)) {
            String[] splitArry = dirPrefix.split("/");
            System.out.println(Arrays.toString(splitArry));
            String tmp = "";
            for(int i = 0; i < splitArry.length; i++) {
                if(StringUtils.isEmpty(splitArry[i])) {
                    continue;
                }
                tmp += "/" + splitArry[i];
                if(!zkClient.exists(tmp)) {
                    zkClient.create(tmp,StringUtils.EMPTY, CreateMode.PERSISTENT);
                }
            }
        }

    }

    private void createEmpNode() {
        String ephNode = dirPrefix + "/";
        String selfNode = null;
        System.out.println(ephNode);
        if(StringUtils.isNotEmpty(ephNode)) {
            selfNode = zkClient.createEphemeralSequential(ephNode,"");
        }
        if(selfNode != null) {
            selfNode = selfNode.substring(selfNode.lastIndexOf("/") + 1);
        }
        System.out.println(selfNode);
        checkMin(selfNode);
    }

    private void checkMin( String selfNode) {
        if(StringUtils.isNotEmpty(dirPrefix)) {
           List<String> children =  zkClient.getChildren(dirPrefix);
            System.out.println(children);
            Collections.sort(children);
            String first  = children.get(0);

            if(selfNode.equals(first)) {
                success(selfNode);
            }else {
                fail(selfNode);
            }
        }
    }

    private void success(String selfNode) {
        System.out.println(Thread.currentThread().getName() + ": success");
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String childNode = dirPrefix +"/" + selfNode;
        if(zkClient.exists(childNode)) {
           zkClient.delete(childNode);
           Lock lock = new ReentrantLock();
           lock.newCondition();
           /*todo 其他节点应该订阅该节点事件
                其他节点在没有获到锁之前应该等待
            */
        }
    }

    private void fail(final String failNode) {
        System.out.println(Thread.currentThread().getName() + ":fail");

        // 接收到通知后，再次抢锁
        zkClient.subscribeChildChanges(dirPrefix, new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> list) throws Exception {
                System.out.println("回调:"+s);
                System.out.println("回调列表：" + list);
                checkMin(failNode);
            }
        });
    }



    public static void main(String[] args) {

        for(int i = 0; i < 5; i++)
        new Thread(new Runnable() {
            @Override
            public void run() {
                DistributeLock dl = new DistributeLock();
                dl.getLock();
            }
        }).start();

    }
}
